package com.yanggu.flink.datastream_api.state.operatorstate

import org.apache.flink.api.common.state.{MapStateDescriptor, ValueStateDescriptor}
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object BroadcastStateDemo02 {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val clickDataStream = env.fromElements(
      ("Alice", "login"),
      ("Alice", "buy")
    )

    val patternDataStream = env.fromElements(
      ("login", "pay"),
      ("login", "buy")
    )

    val mapStateDescriptor = new MapStateDescriptor[Void, (String, String)]("mapState", classOf[Void], classOf[(String, String)])

    val broadcastDataStream = patternDataStream.broadcast(mapStateDescriptor)

    clickDataStream
      .keyBy(data => data._1)
      .connect(broadcastDataStream)
      .process(new PatternEvaluator)
      .print("test>>>>>")

    env.execute()

  }

}

class PatternEvaluator extends KeyedBroadcastProcessFunction[String, (String, String), (String, String), String] {

  private val mapStateDescriptor =
    new MapStateDescriptor[Void, (String, String)]("mapState", classOf[Void], classOf[(String, String)])

  private lazy val lastClickDataState = {
    val clickDataDesc = new ValueStateDescriptor[String]("clickData", classOf[String])
    getRuntimeContext.getState(clickDataDesc)
  }

  override def processElement(value: (String, String),
                              ctx: KeyedBroadcastProcessFunction[String, (String, String), (String, String), String]#ReadOnlyContext,
                              out: Collector[String]): Unit = {
    val mapState = ctx.getBroadcastState(mapStateDescriptor)

    val tuple = mapState.get(null)
    if (tuple == null) {
      return
    }

    val lastClickData = lastClickDataState.value()

    if (lastClickData != null && tuple._1.equals(lastClickData) && tuple._2.equals(value._2)) {
      out.collect(s"${ctx.getCurrentKey}符合模式: ${tuple}")
    }

    lastClickDataState.update(value._2)
  }

  override def processBroadcastElement(value: (String, String),
                                       ctx: KeyedBroadcastProcessFunction[String, (String, String), (String, String), String]#Context,
                                       out: Collector[String]): Unit = {
    val mapState = ctx.getBroadcastState(mapStateDescriptor)
    mapState.put(null, value)
  }

}
